package logger

import (
	"context"
	"encoding/json"
	"log"

	"github.com/segmentio/kafka-go"

	"mychat/internal/conf"
)

type kafkaLogger struct {
	baselogger
	client *kafka.Conn
}

func NewKafkaLogger(lc *conf.Logger) Logger {
	kc := lc.Kafka
	conn, err := kafka.DialLeader(context.Background(), "tcp",
		kc.Addr, kc.Topic, kc.Partition)
	if err != nil {
		panic(err)
	}

	level := convertFromString(lc.Level)
	l := &kafkaLogger{
		baselogger: baselogger{
			tag:      LoggerTagFluentd,
			logLevel: level,
		},
		client: conn,
	}
	l.Logger = l
	return l
}

func (l *kafkaLogger) WithCustomTag(tag LoggerTag) Logger {
	nl := *l
	nl.tag = tag
	nl.Logger = &nl
	return &nl
}

func (l *kafkaLogger) WithCustomStrTag(tagstr string) Logger {
	return l.WithCustomTag(LoggerTag(tagstr))
}

func (l *kafkaLogger) LogEntry(logEntry map[string]any) {
	if l.client == nil {
		return
	}
	valjs, err := json.Marshal(logEntry)
	if err != nil {
		log.Printf("kafka marshal entrys error||err=%v", err)
		return
	}

	_, err = l.client.WriteMessages(kafka.Message{Value: valjs})
	if err != nil {
		log.Printf("kafka write message error||err=%v", err)
	}
}
